package com.shujia;

import javafx.scene.control.Tab;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.List;

public class Demo2API {
    Connection conn = null;
    Admin admin = null;

    @Before
    public void init() {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        try {
            conn = ConnectionFactory.createConnection(conf);
            admin = conn.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 创建表
    @Test
    public void createTable() throws IOException {
        TableName test1 = TableName.valueOf("test1");
        // 判断表是否已经创建 是则删除
        if (admin.tableExists(test1)) {
            admin.disableTable(test1);
            admin.deleteTable(test1);
        }
        HTableDescriptor test1d = new HTableDescriptor(test1);
        HColumnDescriptor cf1 = new HColumnDescriptor("cf1");
        cf1.setMaxVersions(3);
        test1d.addFamily(cf1);
        admin.createTable(test1d);
    }

    // 列出所有的表
    @Test
    public void listTable() throws IOException {
        TableName[] tableNames = admin.listTableNames();
        for (TableName tableName : tableNames) {
            System.out.println(tableName);
        }
    }

    // 删除表
    @Test
    public void dropTable() throws IOException {
        // 删除表之前得先关闭表
        TableName table = TableName.valueOf("test1");
        admin.disableTable(table);
        admin.deleteTable(table);
    }

    // 修改表
    @Test
    public void modifyTable() throws IOException {
        TableName test1 = TableName.valueOf("test1");
        // 通过Admin对象去创建HTableDescriptor对象，如果直接通过new方式创建 则获取不到表本身有的结构
        HTableDescriptor hTableDescriptor = admin.getTableDescriptor(test1);

        // 添加新的列簇cf2
        hTableDescriptor.addFamily(new HColumnDescriptor("cf2"));
        admin.modifyTable(test1, hTableDescriptor);
    }

    // 插入一条数据
    @Test
    public void put() throws IOException {
        // 获取待插入数据的表
        Table test1 = conn.getTable(TableName.valueOf("test1"));
        Put put = new Put("003".getBytes());
        put.addColumn("cf1".getBytes(), "name".getBytes(), "王五".getBytes());
        put.addColumn("cf2".getBytes(), "clazz".getBytes(), "文科二班".getBytes());
        test1.put(put);
    }

    // 通过rowkey获取一条数据
    @Test
    public void get() throws IOException {
        // 连接表
        Table test1 = conn.getTable(TableName.valueOf("test1"));
        Get get = new Get("001".getBytes());
        Result rs = test1.get(get);
        byte[] cf1 = "cf1".getBytes();
        byte[] cf2 = "cf2".getBytes();
        String name = Bytes.toString(rs.getValue(cf1, "name".getBytes()));
        int age = Bytes.toInt(rs.getValue(cf1, "age".getBytes()));
        String gender = Bytes.toString(rs.getValue(cf1, "gender".getBytes()));
        String clazz = Bytes.toString(rs.getValue(cf2, "clazz".getBytes()));
        System.out.println(name + "," + age + "," + gender + "," + clazz);

    }

    // 指定rk的范围进行scan
    @Test
    public void scan() throws IOException {
        // 连接表
        Table test1 = conn.getTable(TableName.valueOf("test1"));
        Scan scan = new Scan();
        // 包含StartRow 不包含StopRow
        scan.withStartRow("001".getBytes());
        scan.withStopRow("004".getBytes());
        ResultScanner scanner = test1.getScanner(scan);
        for (Result rs : scanner) {
            // 获取rowkey
            String rowkey = Bytes.toString(rs.getRow());
            if ("001".equals(rowkey) || "002".equals(rowkey)) {
                byte[] cf1 = "cf1".getBytes();
                byte[] cf2 = "cf2".getBytes();
                String name = Bytes.toString(rs.getValue(cf1, "name".getBytes()));
                int age = Bytes.toInt(rs.getValue(cf1, "age".getBytes()));
                String gender = Bytes.toString(rs.getValue(cf1, "gender".getBytes()));
                String clazz = Bytes.toString(rs.getValue(cf2, "clazz".getBytes()));
                System.out.println(name + "," + age + "," + gender + "," + clazz);
            } else if ("003".equals(rowkey)) {
                byte[] cf1 = "cf1".getBytes();
                byte[] cf2 = "cf2".getBytes();
                String name = Bytes.toString(rs.getValue(cf1, "name".getBytes()));
                String clazz = Bytes.toString(rs.getValue(cf2, "clazz".getBytes()));
                System.out.println(name + "," + clazz);
            }

        }
    }

    // 使用cellUtil获取数据
    @Test
    public void cellUtil() throws IOException {
        // 连接表
        Table test1 = conn.getTable(TableName.valueOf("test1"));
        Scan scan = new Scan();
        // 包含StartRow 不包含StopRow
        scan.withStartRow("001".getBytes());
        scan.withStopRow("004".getBytes());
        ResultScanner scanner = test1.getScanner(scan);
        for (Result rs : scanner) {
            // 获取当前rowkey对应的这条数据的所有的cell
            List<Cell> columnCells = rs.listCells();
            for (Cell cell : columnCells) {
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String qf = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = null;
                if ("age".equals(qf)) {
                    value = Bytes.toInt(CellUtil.cloneValue(cell)) + "";
                } else {
                    value = Bytes.toString(CellUtil.cloneValue(cell));
                }

                System.out.print("rowkey:" + rowkey + ",columnFamily:" + cf + ",Qualifier:" + qf + ",value:" + value + " ");
            }
            System.out.println();

        }
    }

    @Test
    // 将学生信息数据逐条写入HBase
    public void putAll() throws IOException {
        // 创建学生表
        TableName students = TableName.valueOf("students");
        if (!admin.tableExists(students)) {
            HTableDescriptor hTableDescriptor = new HTableDescriptor(students);
            HColumnDescriptor info = new HColumnDescriptor("info");
            hTableDescriptor.addFamily(info);
            admin.createTable(hTableDescriptor);
        }
        Table table = conn.getTable(students);


        BufferedReader bf = new BufferedReader(new FileReader("data/students.txt"));
        String line = null;

        while ((line = bf.readLine()) != null) {
            String[] splits = line.split(",");
            String id = splits[0];
            String name = splits[1];
            String age = splits[2];
            String gender = splits[3];
            String clazz = splits[4];
            Put put = new Put(id.getBytes());
            byte[] info = "info".getBytes();
            put.addColumn(info, "name".getBytes(), name.getBytes());
            put.addColumn(info, "age".getBytes(), age.getBytes());
            put.addColumn(info, "gender".getBytes(), gender.getBytes());
            put.addColumn(info, "clazz".getBytes(), clazz.getBytes());
            table.put(put);
        }
    }

    // 根据rowkey删除数据
    @Test
    public void deleteByRK() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));
        Delete delete = new Delete("1500100069".getBytes());
        students.delete(delete);
    }

    // 获取所有的学生信息
    @Test
    public void getALL() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));
        Scan scan = new Scan();
        ResultScanner scanner = students.getScanner(scan);
        for (Result result : scanner) {
            // result表示一条数据
            String rowkey = Bytes.toString(result.getRow());
            System.out.print(rowkey + " ");
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.print(value + " ");
            }
            System.out.println();
        }
    }


    @After
    public void close() {
        try {
            admin.close();
            conn.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
